package com.guqj.mq.activitemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class ActiveMQFactory {

	// null
	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
	// null
	private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
	// failover://tcp://localhost:61616
	private static final String BOKER_URL = ActiveMQConnection.DEFAULT_BROKER_URL;
	
	/**
	 * 创建Session
	 * 
	 * @return
	 * @throws JMSException
	 */
	public static Connection createConnection() throws JMSException {
		// 实例化连接工厂
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
				USERNAME, PASSWORD, BOKER_URL);
		// 通过连接工厂获取连接
		Connection connection = connectionFactory.createConnection();
		// 启动连接
		connection.start();
		return connection;
	}
	
	/**
	 * 获取消息消费者
	 * 
	 * @param session
	 * @param queue
	 * @return
	 * @throws JMSException
	 */
	public static MessageConsumer createConsumer(Session session, String queue, String type)
			throws JMSException {
		// 创建Destination
		Destination destination = session.createQueue(queue);
		if("TOPIC".equals(type)){
			destination = session.createTopic(queue);
		}
		// 创建消费者
		MessageConsumer messageConsumer = session.createConsumer(destination);
		return messageConsumer;
	}
	
	/**
	 * 消息监听
	 * 
	 * @param messageProducer
	 * @param message
	 * @throws JMSException
	 */
	public static void listenerMessage(final MessageConsumer messageConsumer, final String from)
			throws JMSException {
		messageConsumer.setMessageListener(new MessageListener(){
			@Override
			public void onMessage(Message message) {
				TextMessage tm = (TextMessage) message;  
                try {  
                    System.out.println("Received message From "+ from +": " + tm.getText());  
                } catch (JMSException e) {  
                    e.printStackTrace();  
                }
			}
	    });
	}
	
	/**
	 * 获取消息生产者
	 * 
	 * @param session
	 * @param queue
	 * @return
	 * @throws JMSException
	 */
	public static MessageProducer createProducer(Session session, String queue)
			throws JMSException {
		// 创建队列
		Destination destination = session.createQueue(queue);
		// 创建生产者
		MessageProducer messageProducer = session.createProducer(destination);
		return messageProducer;
	}
	
	/**
	 * 发生消息
	 * 
	 * @param session
	 * @param messageProducer
	 * @param message
	 * @throws JMSException
	 */
	public static void sendMessage(Session session, MessageProducer messageProducer,
			String message) throws JMSException {
		TextMessage textMessage = session.createTextMessage(message);
		messageProducer.send(textMessage);
		System.out.println("发生消息：" + textMessage.getText());
	}
	
	public static void closeSource(Connection connection, Session session){
		try {
			if (null != connection) {
				connection.close();
			}
			if (null != session) {
				session.close();
			}
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}
}
